package com.xyz.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * MQTT发送客户端
 */
@Component
public class MqttSendClient {

    private static final Logger logger = LoggerFactory.getLogger(MqttSendClient.class);

    @Autowired
    private MqttSendCallBack mqttSendCallBack;

    @Autowired
    private MqttConfig mqtt;

    public static MqttClient client;

    private static MqttClient getClient() {
        return client;
    }

    private static void setClient(MqttClient client) {
        MqttSendClient.client = client;
    }

    public void connect() {
        MqttClient client;
        try {
            // String uuid = UUID.randomUUID().toString().replaceAll("-", "");
            client = new MqttClient(mqtt.getHostUrl(), mqtt.getClientId(), new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(mqtt.getUsername());
            options.setPassword(mqtt.getPassword().toCharArray());
            options.setConnectionTimeout(mqtt.getTimeout());
            options.setKeepAliveInterval(mqtt.getKeepAlive());
            options.setAutomaticReconnect(mqtt.getReconnect());
            options.setCleanSession(mqtt.getCleanSession());
            MqttSendClient.setClient(client);
            // 设置回调
            client.setCallback(mqttSendCallBack);
            client.connect(options);
        } catch (Exception e) {
            logger.warn("MqttSendClient connect error,message:{}", e.getMessage());
        }
    }

    /**
     * 发布消息
     *
     * @param retained 是否保留
     * @param topic    主题
     * @param content  消息内容
     */
    public void publish(boolean retained, String topic, String content) {
        MqttMessage message = new MqttMessage();
        message.setQos(mqtt.getQos());
        message.setRetained(retained);
        message.setPayload(content.getBytes());
        message.setRetained(true);
        message.setPayload(content.getBytes(StandardCharsets.UTF_8));
        try {
            client.publish(topic, message);
        } catch (MqttException e) {
            logger.warn("MqttSendClient publish error,message:{}", e.getMessage());
        }
    }

    /**
     * 订阅某个主题
     */
    public void subscribe(String topic, int qos) {
        logger.warn("========================【开始订阅主题:" + topic + "】========================");
        try {
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            logger.warn("MqttAcceptClient subscribe warn,message:{}", e.getMessage());
        }
    }

    /**
     * 取消订阅某个主题
     */
    public void unsubscribe(String topic) {
        logger.warn("========================【取消订阅主题:" + topic + "】========================");
        try {
            client.unsubscribe(topic);
        } catch (MqttException e) {
            logger.warn("MqttAcceptClient unsubscribe warn,message:{}", e.getMessage());
        }
    }

    /**
     * 重新连接
     */
    public void reconnection() {
        try {
            client.connect();
        } catch (MqttException e) {
            logger.warn("MqttAcceptClient reconnection warn,message:{}", e.getMessage());
        }
    }

    /**
     * 关闭连接
     */
    public static void disconnect(MqttClient mqttClient) {
        try {
            if (mqttClient != null)
                mqttClient.disconnect();
        } catch (MqttException e) {
            logger.warn("MqttSendClient disconnect error,message:{}", e.getMessage());
        }
    }

    /**
     * 释放资源
     */
    public static void close(MqttClient mqttClient) {
        try {
            if (mqttClient != null)
                mqttClient.close();
        } catch (MqttException e) {
            logger.warn("MqttSendClient close error,message:{}", e.getMessage());
        }
    }

}
